package io.kiki.sba.registry.server.clientInterface.bootstrap;


import io.kiki.sba.registry.task.MetricsableThreadPoolExecutor;
import io.kiki.sba.registry.util.NamedThreadFactory;
import io.kiki.sba.registry.util.OsUtils;
import lombok.Getter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;


public class ExecutorManager {

    private static final Logger logger = LoggerFactory.getLogger(ExecutorManager.class);
    private static final String ACCESS_DATA_EXECUTOR = "AccessDataExecutor";
    private static final String ACCESS_SUB_EXECUTOR = "AccessSubExecutor";
    private static final String DATA_CHANGE_REQUEST_EXECUTOR = "DataChangeExecutor";
    private static final String DATA_SLOT_SYNC_REQUEST_EXECUTOR = "SlotSyncExecutor";
    private static final String ACCESS_METADATA_EXECUTOR = "AccessMetadataExecutor";
    private static final String CONSOLE_EXECUTOR = "ConsoleExecutor";
    private static final String ZONE_SDK_EXECUTOR = "ZoneSdkExecutor";
    private static final String CLIENT_MANAGER_CHECK_EXECUTOR = "ClientManagerCheckExecutor";
    private static final String APP_REVISION_REGISTER_EXECUTOR = "AppRevisionRegisterExecutor";
    private static final String SCAN_EXECUTOR = "ScanExecutor";
    private final ScheduledThreadPoolExecutor scheduledThreadPoolExecutor;
    @Getter
    private final ThreadPoolExecutor accessDataExecutor;
    @Getter
    private final ThreadPoolExecutor accessSubExecutor;
    @Getter
    private final ThreadPoolExecutor dataChangeRequestExecutor;
    @Getter
    private final ThreadPoolExecutor dataSlotSyncRequestExecutor;
    @Getter
    private final ThreadPoolExecutor accessMetadataExecutor;
    @Getter
    private final ThreadPoolExecutor consoleExecutor;
    @Getter
    private final ThreadPoolExecutor zoneSdkExecutor;
    @Getter
    private final ThreadPoolExecutor clientManagerCheckExecutor;
    @Getter
    private final ThreadPoolExecutor scanExecutor;
    @Getter
    private final ThreadPoolExecutor appRevisionRegisterExecutor;
    @Getter
    private Map<String, ThreadPoolExecutor> reportExecutors = new HashMap<>();

    public ExecutorManager(Config config) {
        scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(config.getSessionSchedulerPoolSize(), new NamedThreadFactory("SessionScheduler"));

        accessDataExecutor = reportExecutors.computeIfAbsent(ACCESS_DATA_EXECUTOR, k -> new MetricsableThreadPoolExecutor(ACCESS_DATA_EXECUTOR, config.getAccessDataExecutorPoolSize(), config.getAccessDataExecutorPoolSize(), 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(config.getAccessDataExecutorQueueSize()), new NamedThreadFactory(ACCESS_DATA_EXECUTOR, true), (r, executor) -> {
            String msg = String.format("Task(%s) %s rejected from %s, just ignore it to let client timeout.", r.getClass(), r, executor);
            logger.error(msg);
        }));

        accessSubExecutor = reportExecutors.computeIfAbsent(ACCESS_SUB_EXECUTOR, k -> new MetricsableThreadPoolExecutor(ACCESS_SUB_EXECUTOR, config.getAccessSubDataExecutorPoolSize(), config.getAccessSubDataExecutorPoolSize(), 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(config.getAccessSubDataExecutorQueueSize()), new NamedThreadFactory(ACCESS_SUB_EXECUTOR, true), (r, executor) -> {
            String msg = String.format("Task(%s) %s rejected from %s, just ignore it to let client timeout.", r.getClass(), r, executor);
            logger.error(msg);
        }));
        dataChangeRequestExecutor = reportExecutors.computeIfAbsent(DATA_CHANGE_REQUEST_EXECUTOR, k -> new MetricsableThreadPoolExecutor(DATA_CHANGE_REQUEST_EXECUTOR, config.getDataChangeExecutorPoolSize(), config.getDataChangeExecutorPoolSize(), 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(config.getDataChangeExecutorQueueSize()), new NamedThreadFactory(DATA_CHANGE_REQUEST_EXECUTOR, true)));

        dataSlotSyncRequestExecutor = reportExecutors.computeIfAbsent(DATA_SLOT_SYNC_REQUEST_EXECUTOR, k -> new MetricsableThreadPoolExecutor(DATA_SLOT_SYNC_REQUEST_EXECUTOR, config.getSlotSyncWorkerSize(), config.getSlotSyncWorkerSize(), 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(config.getSlotSyncMaxBufferSize()), new NamedThreadFactory(DATA_SLOT_SYNC_REQUEST_EXECUTOR, true)));
        accessMetadataExecutor = reportExecutors.computeIfAbsent(ACCESS_METADATA_EXECUTOR, k -> new MetricsableThreadPoolExecutor(ACCESS_METADATA_EXECUTOR, config.getAccessMetadataWorkerSize(), config.getAccessMetadataWorkerSize(), 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(config.getAccessMetadataMaxBufferSize()), new NamedThreadFactory(ACCESS_METADATA_EXECUTOR, true)));

        consoleExecutor = reportExecutors.computeIfAbsent(CONSOLE_EXECUTOR, k -> new MetricsableThreadPoolExecutor(CONSOLE_EXECUTOR, config.getConsoleExecutorPoolSize(), config.getConsoleExecutorPoolSize(), 60L, TimeUnit.SECONDS, new LinkedBlockingQueue(config.getConsoleExecutorQueueSize()), new NamedThreadFactory(CONSOLE_EXECUTOR, true)));

        zoneSdkExecutor = reportExecutors.computeIfAbsent(ZONE_SDK_EXECUTOR, k -> MetricsableThreadPoolExecutor.newExecutor(ZONE_SDK_EXECUTOR, OsUtils.getCpuCount() * 5, 100, new ThreadPoolExecutor.CallerRunsPolicy()));

        clientManagerCheckExecutor = reportExecutors.computeIfAbsent(CLIENT_MANAGER_CHECK_EXECUTOR, k -> MetricsableThreadPoolExecutor.newExecutor(CLIENT_MANAGER_CHECK_EXECUTOR, OsUtils.getCpuCount() * 5, 100, new ThreadPoolExecutor.CallerRunsPolicy()));

        appRevisionRegisterExecutor = reportExecutors.computeIfAbsent(APP_REVISION_REGISTER_EXECUTOR, k -> new MetricsableThreadPoolExecutor(APP_REVISION_REGISTER_EXECUTOR, config.getMetadataRegisterExecutorPoolSize(), config.getMetadataRegisterExecutorPoolSize(), 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(config.getMetadataRegisterExecutorQueueSize()), new NamedThreadFactory(APP_REVISION_REGISTER_EXECUTOR, true), new ThreadPoolExecutor.CallerRunsPolicy()));

        scanExecutor = reportExecutors.computeIfAbsent(SCAN_EXECUTOR, k -> new MetricsableThreadPoolExecutor(SCAN_EXECUTOR, config.getScanExecutorPoolSize(), config.getScanExecutorPoolSize(), 60, TimeUnit.SECONDS, new ArrayBlockingQueue<>(config.getScanExecutorQueueSize()), new NamedThreadFactory(SCAN_EXECUTOR, true), new ThreadPoolExecutor.CallerRunsPolicy()));
    }

    public void startScheduler() {
    }

    public void stopScheduler() {
        scheduledThreadPoolExecutor.shutdown();

        accessDataExecutor.shutdown();

        dataChangeRequestExecutor.shutdown();

        dataSlotSyncRequestExecutor.shutdown();
    }


}
